package org.raymond.iworks.study.mq.rabbitmq.classic.notlose;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.function.BooleanSupplier;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class DurableTask {
    public static void main(String[] args) throws Exception {
        String message = "hello...";
        try(Connection connection = DurableWorker.connection();
            Channel channel = connection.createChannel()){

            String queueName = UUID.randomUUID().toString();
            boolean durable = false;
            channel.queueDeclare(queueName, durable, false, true, null);
            channel.exchangeDeclare(DurableWorker.EXCHANGE_NAME, "topic");

            channel.confirmSelect();

            ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap();

            ConfirmCallback cleanWhenAckCallback = (sequenceNumber, multiple)->{
                log.info("ack callback seqNo:{}", sequenceNumber);
                if(multiple){
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(
                            sequenceNumber, true
                    );
                    confirmed.clear();
                }else{
                    outstandingConfirms.remove(sequenceNumber);
                }
            };

            ConfirmCallback nackCallback = (sequenceNumber, multiple)->{
                String body = outstandingConfirms.get(sequenceNumber);
                log.error("Message with body {} has been nack-ed. Sequence number: {}, multiple: {}",
                        body, sequenceNumber, multiple);
                cleanWhenAckCallback.handle(sequenceNumber, multiple);
            };

            channel.addConfirmListener(cleanWhenAckCallback, nackCallback);

            long start = System.nanoTime();
            // 异步发布确认的步骤-3. 在发布消息之前跟踪publishSeqNo
            long seqNo = channel.getNextPublishSeqNo();
            log.info("bind seqNo:{}", seqNo);
            outstandingConfirms.put(seqNo, message);

            channel.basicPublish(DurableWorker.EXCHANGE_NAME, DurableWorker.BINDING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));

            if(!waitUnit(Duration.ofSeconds(60), outstandingConfirms::isEmpty)){
                throw new IllegalStateException("All messages could not be confirmed in 60 seconds");
            }
            long end = System.nanoTime();
            log.info("Published message {}, in batch in {}", message, Duration.ofNanos(end - start).toMillis());
        }
    }

    private static boolean waitUnit(Duration timeout, BooleanSupplier condition) throws InterruptedException {
        int waited = 0;
        while(!condition.getAsBoolean() && waited < timeout.toMillis()){
            Thread.sleep(100L);
            waited += 100;
        }
        return condition.getAsBoolean();
    }
}
